package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

public class StatBoltMin extends AbstractBolt{

	private static final long serialVersionUID = 1L;

	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	private HashMap<String,Long> upStreampacket = new HashMap<String,Long>();
	
	private HashMap<String,Long> downStreamoctets = new HashMap<String,Long>();

	private HashMap<String,Long> downStreampacket = new HashMap<String,Long>();
	
	private HashMap<String,String> mapIPQQ = new HashMap<String,String>();
		
//	private HashMap<String,Integer> allupStreamoctets = new HashMap<String,Integer>();
//
//	private HashMap<String,Integer> allupStreampacket = new HashMap<String,Integer>();
//	
//	private HashMap<String,Integer> alldownStreamoctets = new HashMap<String,Integer>();
//
//	private HashMap<String,Integer> alldownStreampacket = new HashMap<String,Integer>();
	
	
	@Override
	public void execute(Tuple tuple) {		
		try {
			if(isTickTuple(tuple)) {
				Logger.getLogger(StatBoltMin.class).info(String.format("###DEBUG:UDPFLOOD START EMIT [%s]",tuple.getSourceStreamId()));
				emitCountingData(collector);
				Logger.getLogger(StatBoltMin.class).info(String.format("###DEBUG:UDPFLOOD END EMIT [%s]",tuple.getSourceStreamId()));
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {			
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("StatBoltMin throws an exception!",e);
		}
	}


	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{	
//			SimpleDateFormat sdfm = new SimpleDateFormat("mm");
//			SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH");
//			Date date = new Date();
//			String dateStrM = sdfm.format(date);
//			String dateStrH = sdfh.format(date);
//			int nMin5 = (Integer.parseInt(dateStrM)/5)*5;			
//			String sMin5 = String.format("%02d", nMin5);			
//			String dateStr5Key = dateStrH+sMin5+"00";			
			
			if(tuple.contains("upstreampacket") && tuple.contains("upstreamoctets") && tuple.contains("dnstreampacket") && tuple.contains("dnstreamoctets")){				
				String sip = tuple.getStringByField("sip");	
				if(sip.equals("192.168.3.223"))//魔高一尺，道高一丈
					return;
				
				long uppacket = tuple.getLongByField("upstreampacket");
				long downpacket = tuple.getLongByField("dnstreampacket");	
				long upstreamoctets = tuple.getLongByField("upstreamoctets");				
				long downstreamoctets = tuple.getLongByField("dnstreamoctets");
				
				if(upStreamoctets.containsKey(sip)) {
					upStreamoctets.put(sip,upStreamoctets.get(sip) + upstreamoctets);
				}else{
					upStreamoctets.put(sip,upstreamoctets);
				}
				
				if(downStreamoctets.containsKey(sip)) {
					downStreamoctets.put(sip,downStreamoctets.get(sip) + downstreamoctets);
				}else{
					downStreamoctets.put(sip,downstreamoctets);
				}
				
				if(upStreampacket.containsKey(sip)) {
					upStreampacket.put(sip,upStreampacket.get(sip) + uppacket);
				}else{
					upStreampacket.put(sip,uppacket);
				}
				
				if(downStreampacket.containsKey(sip)) {
					downStreampacket.put(sip,downStreampacket.get(sip) + downpacket);
				}else{
					downStreampacket.put(sip,downpacket);
				}
				
				//获取QQ号
				String domain = tuple.getStringByField("domainname").trim();
				if(domain.contains("qq.com")) {
					String cookies = tuple.getStringByField("cookie").trim();
					//;uin=o2810686021;
					if(cookies.indexOf(";uin=o") > 0) {
						String strQQ = cookies.substring(cookies.indexOf(";uin=o") + 6);
						strQQ = strQQ.substring(0,strQQ.indexOf(";"));
						mapIPQQ.put(sip, strQQ);
					}
					
				}
				
//				if(allupStreamoctets.containsKey(dateStr5Key)) {
//					allupStreamoctets.put(dateStr5Key,allupStreamoctets.get(dateStr5Key) + upstreamoctets);
//				}else{
//					allupStreamoctets.put(dateStr5Key,upstreamoctets);
//				}
//				
//				if(alldownStreamoctets.containsKey(dateStr5Key)) {
//					alldownStreamoctets.put(dateStr5Key,alldownStreamoctets.get(dateStr5Key) + downstreamoctets);
//				}else{
//					alldownStreamoctets.put(dateStr5Key,downstreamoctets);
//				}
//				
//				if(allupStreampacket.containsKey(dateStr5Key)) {
//					allupStreampacket.put(dateStr5Key,allupStreampacket.get(dateStr5Key) + uppacket);
//				}else{
//					allupStreampacket.put(dateStr5Key,uppacket);
//				}
//				
//				if(alldownStreampacket.containsKey(dateStr5Key)) {
//					alldownStreampacket.put(dateStr5Key,alldownStreampacket.get(dateStr5Key) + downpacket);
//				}else{
//					alldownStreampacket.put(dateStr5Key,downpacket);
//				}

			}
	
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====StatBoltMin_countInLocal=========ERROR=================",ex);
		}
	}


	public void emitCountingData(OutputCollector collector) {
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		
		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH = sdfh.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
		String sMin5 = String.format("%02d", nMin5);
		String sMin30 = String.format("%02d", nMin30);
		String dateStr5 = dateStrH+sMin5 +"00";
		String dateStr30 = dateStrH+sMin30 +"00";
		String rowKeyStat5Min = null;
		
		try {
			
			for(Entry<String, Long> entry : upStreamoctets.entrySet()) {

				String entryKey = entry.getKey();
				long upoct = entry.getValue();
				long uppkg = upStreampacket.get(entryKey);
				long dnoct = downStreamoctets.get(entryKey);
				long dnpkg = downStreampacket.get(entryKey);
				//rowKeyStat5Min = dateStr5 + "_" + entryKey ;	
				rowKeyStat5Min = entryKey + "_" + dateStr5;	
				
				hbaseInstance.Add("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf", "REPORTTIME", dateStr5);
				hbaseInstance.Add("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf", "IP", entryKey);
				hbaseInstance.incerment("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf:UPOCTETS", Long.valueOf(upoct));
				hbaseInstance.incerment("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf:UPPACKET", Long.valueOf(uppkg));
				hbaseInstance.incerment("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf:DNOCTETS", Long.valueOf(dnoct));							
				hbaseInstance.incerment("METIS_FLOW_STAT_5MIN", rowKeyStat5Min, "cf:DNPACKET", Long.valueOf(dnpkg));
								
				hbaseInstance.Add("METIS_ALL_FLOW_STAT_5MIN", dateStr5, "cf", "REPORTTIME", dateStr5);				
				hbaseInstance.incerment("METIS_ALL_FLOW_STAT_5MIN", dateStr5, "cf:UPOCTETS", Long.valueOf(upoct));
				hbaseInstance.incerment("METIS_ALL_FLOW_STAT_5MIN", dateStr5, "cf:UPPACKET", Long.valueOf(uppkg));
				hbaseInstance.incerment("METIS_ALL_FLOW_STAT_5MIN", dateStr5, "cf:DNOCTETS", Long.valueOf(dnoct));							
				hbaseInstance.incerment("METIS_ALL_FLOW_STAT_5MIN", dateStr5, "cf:DNPACKET", Long.valueOf(dnpkg));
				String qq = "";
				if(mapIPQQ.containsKey(entry.getKey())) {
					qq = mapIPQQ.get(entry.getKey());
				}
				
				
				collector.emit(new Values(entryKey,dateStr30,upoct,uppkg,dnoct,dnpkg, qq));
			}
			
			
			upStreamoctets.clear();
			upStreampacket.clear();
			downStreamoctets.clear();
			downStreampacket.clear();
		
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("StatBoltMin Exception",e);;
		}
	}



	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
		declarer.declare(new Fields("sip","dateStr30","upstreamoctets","upstreampacket","dnstreamoctets","dnstreampacket","qq"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300); //5分钟统计一次
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub

	}

}